import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 任务四：优惠券使用影响因素分析 - 折扣率对消费行为的影响
 * 分析不同折扣率下的优惠券使用情况
 * 输出格式：<折扣率> TAB <领取数> TAB <使用数> TAB <使用率%>
 */
public class Task4DiscountRateAnalysis {

    public static class DiscountMapper extends Mapper<Object, Text, Text, Text> {
        private Text discountRate = new Text();
        private Text usageInfo = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String[] fields = value.toString().split(",", -1);
            
            if (fields.length < 7 || fields[0].equals("User_id")) {
                return;
            }

            String couponId = fields[2];      // Coupon_id
            String discount = fields[3];      // Discount_rate
            String date = fields[6];          // Date

            // 只统计有优惠券的记录
            if (couponId == null || couponId.equals("null") || couponId.isEmpty()) {
                return;
            }
            if (discount == null || discount.equals("null") || discount.isEmpty()) {
                return;
            }

            // 解析折扣率
            String discountCategory = parseDiscountRate(discount);
            if (discountCategory == null) {
                return;
            }

            discountRate.set(discountCategory);

            // 判断是否使用：有Date表示已使用，无Date表示仅领取未使用
            boolean used = (date != null && !date.equals("null") && !date.isEmpty());

            if (used) {
                usageInfo.set("used:1");
            } else {
                usageInfo.set("received:1");
            }
            context.write(discountRate, usageInfo);
        }

        // 解析折扣率并分类（统一使用1位小数）
        private String parseDiscountRate(String discount) {
            if (discount.contains(":")) {
                // 满减类型，如 "30:5" 表示满30减5
                String[] parts = discount.split(":");
                try {
                    double full = Double.parseDouble(parts[0]);
                    double reduce = Double.parseDouble(parts[1]);
                    double rate = (full - reduce) / full;
                    return String.format("满减%.1f折", rate * 10);
                } catch (NumberFormatException e) {
                    return null;
                }
            } else {
                // 直接折扣，如 "0.8" 表示8折
                try {
                    double rate = Double.parseDouble(discount);
                    if (rate > 0 && rate <= 1) {
                        return String.format("%.1f折", rate * 10);
                    }
                } catch (NumberFormatException e) {
                    return null;
                }
            }
            return null;
        }
    }

    public static class DiscountReducer extends Reducer<Text, Text, Text, Text> {
        private Text result = new Text();

        public void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            int receivedCount = 0;
            int usedCount = 0;

            for (Text val : values) {
                String[] parts = val.toString().split(":");
                if (parts.length == 2) {
                    int count = Integer.parseInt(parts[1]);
                    if (parts[0].equals("received")) {
                        receivedCount += count;
                    } else if (parts[0].equals("used")) {
                        usedCount += count;
                    }
                }
            }

            int totalReceived = receivedCount + usedCount;
            double usageRate = totalReceived > 0 ? (double) usedCount / totalReceived * 100 : 0;

            result.set(totalReceived + "\t" + usedCount + "\t" + String.format("%.2f%%", usageRate));
            context.write(key, result);
        }
    }

    // 排序Job
    public static class SortMapper extends Mapper<Object, Text, DoubleWritable, Text> {
        private DoubleWritable usageRate = new DoubleWritable();
        private Text info = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String[] parts = value.toString().split("\t");
            if (parts.length >= 4) {
                try {
                    String rateStr = parts[3].replace("%", "");
                    double rate = Double.parseDouble(rateStr);
                    usageRate.set(-rate); // 负数实现降序
                    info.set(parts[0] + "\t" + parts[1] + "\t" + parts[2] + "\t" + parts[3]);
                    context.write(usageRate, info);
                } catch (NumberFormatException e) {
                    // 忽略
                }
            }
        }
    }

    public static class SortReducer extends Reducer<DoubleWritable, Text, Text, Text> {
        private Text emptyKey = new Text("");

        public void reduce(DoubleWritable key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            for (Text val : values) {
                context.write(emptyKey, val);
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        
        // Job 1: 统计各折扣率的使用情况
        Job job1 = Job.getInstance(conf, "Task4 Discount Rate Analysis - Step1");
        job1.setJarByClass(Task4DiscountRateAnalysis.class);
        job1.setMapperClass(DiscountMapper.class);
        job1.setReducerClass(DiscountReducer.class);
        job1.setOutputKeyClass(Text.class);
        job1.setOutputValueClass(Text.class);
        
        FileInputFormat.addInputPath(job1, new Path(args[0]));
        Path tempOutput = new Path(args[1] + "_temp");
        FileOutputFormat.setOutputPath(job1, tempOutput);
        
        if (!job1.waitForCompletion(true)) {
            System.exit(1);
        }

        // Job 2: 按使用率排序
        Job job2 = Job.getInstance(conf, "Task4 Discount Rate Analysis - Step2");
        job2.setJarByClass(Task4DiscountRateAnalysis.class);
        job2.setMapperClass(SortMapper.class);
        job2.setReducerClass(SortReducer.class);
        job2.setMapOutputKeyClass(DoubleWritable.class);
        job2.setMapOutputValueClass(Text.class);
        job2.setOutputKeyClass(Text.class);
        job2.setOutputValueClass(Text.class);
        
        FileInputFormat.addInputPath(job2, tempOutput);
        FileOutputFormat.setOutputPath(job2, new Path(args[1]));
        
        System.exit(job2.waitForCompletion(true) ? 0 : 1);
    }
}
